package com.atbeijing.D06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 双流intervalJoin
 * 仅支持事件时间
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple3<String, String, Long>> clickStream = env
                .fromElements(
                        Tuple3.of("user-1", "click", 20 * 60 * 1000L),
                        Tuple3.of("user-1", "click", 50 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                                        return element.f2;
                                    }
                                })
                );

        SingleOutputStreamOperator<Tuple3<String, String, Long>> browseStream = env
                .fromElements(
                        Tuple3.of("user-1", "browse", 10 * 60 * 1000-1L),
                        Tuple3.of("user-1", "browse", 10 * 60 * 1000L),
                        Tuple3.of("user-1", "browse", 16 * 60 * 1000L),
                        Tuple3.of("user-1", "browse", 22 * 60 * 1000L),
                        Tuple3.of("user-1", "browse", 22 * 60 * 1000L+1),
                        Tuple3.of("user-1", "browse", 50 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                                        return element.f2;
                                    }
                                })
                );

        clickStream
                //keyBy的目的时通过两条流的key连接
                .keyBy(r -> r.f0)
                //clickStream的每条数据根据时间戳匹配browseStream中一定时间戳范围的数据,范围由between算子设置,仅支持事件时间
                .intervalJoin(browseStream.keyBy(r -> r.f0))
                //20-10,20+2  匹配区间[10,22]
                .between(Time.minutes(-10), Time.minutes(2))
                //处理匹配到的数据
                .process(new ProcessJoinFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String>() {
                    @Override
                    public void processElement(Tuple3<String, String, Long> left, Tuple3<String, String, Long> right, Context ctx, Collector<String> out) throws Exception {
                        out.collect(left + "=>" + right);
                    }
                })
                .print();

        browseStream
                .keyBy(r -> r.f0)
                .intervalJoin(clickStream.keyBy(r -> r.f0))
                .between(Time.minutes(-2), Time.minutes(10))
                .process(new ProcessJoinFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String>() {
                    @Override
                    public void processElement(Tuple3<String, String, Long> left, Tuple3<String, String, Long> right, Context ctx, Collector<String> out) throws Exception {
                        out.collect(left + "=>>>" + right);
                    }
                })
                .print();

        env.execute();
    }
}
